package com.atguigu.gmall.realtime.app.dws;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.BaseApp;
import com.atguigu.gmall.realtime.bean.TradeProvinceOrderWindow;
import com.atguigu.gmall.realtime.common.Constant;
import com.atguigu.gmall.realtime.function.AsyncDimFunction;
import com.atguigu.gmall.realtime.util.AtguiguUtil;
import com.atguigu.gmall.realtime.util.FlinkSinUtil;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.state.StateTtlConfig;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.AsyncDataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.ProcessWindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.time.Duration;
import java.util.Collections;
import java.util.HashSet;
import java.util.concurrent.TimeUnit;

/**
 * @Author lzc
 * @Date 2023/2/23 08:28
 */
public class Dws_10_DwsTradeProvinceOrderWindow extends BaseApp {
    public static void main(String[] args) {
        new Dws_10_DwsTradeProvinceOrderWindow().init(
            4010,
            2,
            "Dws_10_DwsTradeProvinceOrderWindow",
            Constant.TOPIC_DWD_TRADE_ORDER_DETAIL
        );
    }
    
    @Override
    protected void handle(StreamExecutionEnvironment env, DataStreamSource<String> stream) {
        //1. 封装 pojo
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream = parsePojo(stream);
        // 2. 按照详情 id 去重
        beanStream = distinctByOrderDetailId(beanStream);
        // 3. 开窗聚合
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithoutDim = windowAndAgg(beanStream);
        // 4. 补充维度
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithDim = joinDim(beanStreamWithoutDim);
        beanStreamWithDim.print();
        // 5. 写出到 clickhouse
        writeToClickHouse(beanStreamWithDim);
    
    }
    
    private void writeToClickHouse(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithDim) {
        beanStreamWithDim.addSink(FlinkSinUtil.getClickHouseSink("dws_trade_province_order_window", TradeProvinceOrderWindow.class));
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> joinDim(
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStreamWithoutDim) {
        return AsyncDataStream.unorderedWait(
            beanStreamWithoutDim,
            new AsyncDimFunction<TradeProvinceOrderWindow>() {
                @Override
                public String getTable() {
                    return "dim_base_province";
                }
    
                @Override
                public Object getId(TradeProvinceOrderWindow value) {
                    return value.getProvinceId();
                }
    
                @Override
                public void addDim(TradeProvinceOrderWindow value, JSONObject dim) {
                    value.setProvinceName(dim.getString("NAME"));
                }
            },
            60,
            TimeUnit.SECONDS
        );
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> windowAndAgg(SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
      return  beanStream
            .assignTimestampsAndWatermarks(
                WatermarkStrategy
                    .<TradeProvinceOrderWindow>forBoundedOutOfOrderness(Duration.ofSeconds(3))
                    .withTimestampAssigner((bean, ts) -> bean.getTs())
                    .withIdleness(Duration.ofSeconds(60))
            )
            .keyBy(TradeProvinceOrderWindow::getProvinceId)
            .window(TumblingEventTimeWindows.of(Time.seconds(5)))
            .reduce(
                new ReduceFunction<TradeProvinceOrderWindow>() {
                    @Override
                    public TradeProvinceOrderWindow reduce(TradeProvinceOrderWindow value1,
                                                           TradeProvinceOrderWindow value2) throws Exception {
                        value1.setOrderAmount(value1.getOrderAmount().add(value2.getOrderAmount()));
                        value1.getOrderIdSet().addAll(value2.getOrderIdSet());
                        return value1;
                    }
                },
                new ProcessWindowFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow, String, TimeWindow>() {
                    @Override
                    public void process(String provinceId,
                                        Context ctx,
                                        Iterable<TradeProvinceOrderWindow> elements,
                                        Collector<TradeProvinceOrderWindow> out) throws Exception {
                        TradeProvinceOrderWindow bean = elements.iterator().next();
                        
                        bean.setStt(AtguiguUtil.tsToDateTime(ctx.window().getStart()));
                        bean.setEdt(AtguiguUtil.tsToDateTime(ctx.window().getEnd()));
                        
                        bean.setOrderCount((long) bean.getOrderIdSet().size());
                        bean.setTs(System.currentTimeMillis());
                        
                        out.collect(bean);
                    }
                }
            );
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> distinctByOrderDetailId(
        SingleOutputStreamOperator<TradeProvinceOrderWindow> beanStream) {
        return beanStream
            .keyBy(TradeProvinceOrderWindow::getOrderDetailId)
            .map(new RichMapFunction<TradeProvinceOrderWindow, TradeProvinceOrderWindow>() {
                
                private ValueState<TradeProvinceOrderWindow> lastBeanState;
                
                @Override
                public void open(Configuration parameters) throws Exception {
                    // 给状态设置 ttl
                    StateTtlConfig config = new StateTtlConfig
                        .Builder(org.apache.flink.api.common.time.Time.seconds(10)) // ttl 时间
                        .setUpdateType(StateTtlConfig.UpdateType.OnReadAndWrite) // 当创建写和读的时候,更新 ttl
                        .setStateVisibility(StateTtlConfig.StateVisibility.NeverReturnExpired) // 当过期之后, 如果去获取状态返回 null
                        .build();
                    
                    ValueStateDescriptor<TradeProvinceOrderWindow> desc = new ValueStateDescriptor<>("lastBean", TradeProvinceOrderWindow.class);
                    desc.enableTimeToLive(config);
                    
                    lastBeanState = getRuntimeContext().getState(desc);
                }
                
                @Override
                public TradeProvinceOrderWindow map(TradeProvinceOrderWindow bean) throws Exception {
                    // 当第一条数据来的时候: 1. 把数据存储到状态中 2. 输出
                    
                    // 不是第一条: 1. 对四个指标用新值-状态中的值, 输出  2. 把第二条数据存入到状态
                    
                    TradeProvinceOrderWindow lastBean = lastBeanState.value();
                    if (lastBean != null) {
                        // 计算后的值,存入到 lastBean, 把 lastBean 输出
                        lastBean.setOrderAmount(bean.getOrderAmount().subtract(lastBean.getOrderAmount()));  // 新值 - 旧值
                    }
                    lastBeanState.update(bean);  // 新值存入到状态中
                    return lastBean == null ? bean : lastBean; // 计算后的值
                }
            });
    }
    
    private SingleOutputStreamOperator<TradeProvinceOrderWindow> parsePojo(DataStreamSource<String> stream) {
       return stream.map(new MapFunction<String, TradeProvinceOrderWindow>() {
            @Override
            public TradeProvinceOrderWindow map(String value) throws Exception {
                JSONObject obj = JSON.parseObject(value);
    
    
                return TradeProvinceOrderWindow.builder()
                    .provinceId(obj.getString("province_id"))
                    .orderAmount(obj.getBigDecimal("split_total_amount"))
                    .orderDetailId(obj.getString("id"))
                    .orderIdSet(new HashSet<>(Collections.singleton(obj.getString("order_id"))))
                    .ts(obj.getLong("ts") * 1000)
                    .build();
            }
        });
    }
    
    
}
/*
交易域省份粒度下单各窗口汇总

统计各省份各窗口订单数和订单金额
    
数据源: 下单明细表 1

1. 按照详情 id 去重
        订单金额
2. 订单数
    订单id  详情id
     1      2
     1      3
     1      4
     
        count(distinct(id))
        
   1. 封装 pojo 的是, 同一个订单第一条数据来的时候  订单数=1
        同一个订单其他条数据来的时候 订单数=0
        
   2. 把订单 id 放入到 set 集合中, 最后统计最终结果的时候, 订单数取 set 集合的长度
   
3. 按照省份 id keyBy, 开窗聚和

4. 补充维度信息: 省份名称

5. 写出到 clickhouse 中
 */